package net.bwie.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * 日考2考试
 * @author xuanyu
 * @date 2025/10/16
 */
public class FlinkDemo02 {

	public static void main(String[] args) throws Exception{
		// 1.执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1);

		// todo 5、新建流处理类读取goods_topic主题数据（10分）
		KafkaSource<String> source = KafkaSource.<String>builder()
			.setBootstrapServers("node101:9092,node102:9092,node103:9092")
			.setTopics("goods_topic")
			.setGroupId("my-group-1")
			.setStartingOffsets(OffsetsInitializer.earliest())
			.setValueOnlyDeserializer(new SimpleStringSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

		// todo 6、过滤掉数据中的脏数据（下单个数为0）。（10分）
		SingleOutputStreamOperator<String> stream6 = stream.filter(
			new FilterFunction<String>() {
				@Override
				public boolean filter(String value) throws Exception {
					/*
						1,玩具,玩具手枪,28888.2,12
						商品id,商品类别,商品名称,金额,商品下单个数
					 */
					// 字符串分割
					String[] split = value.split(",");
					// 下单个数
					Integer orderNumber = Integer.valueOf(split[4]);
					// 判断
					return orderNumber != 0;
				}
			}
		);

		// todo 7、根据商品类别分组,求出每个类别商品下单的总个数。（10分）
		// 7-1.提取字段，封装二元组
		SingleOutputStreamOperator<Tuple2<String, Integer>> stream71 = stream6.map(
			new RichMapFunction<String, Tuple2<String, Integer>>() {
				@Override
				public Tuple2<String, Integer> map(String value) throws Exception {
					/*
						1,玩具,玩具手枪,28888.2,12
						商品id,商品类别,商品名称,金额,商品下单个数
					 */
					// 字符串分割
					String[] split = value.split(",");
					// 商品类别、商品下单个数
					return Tuple2.of(split[1], Integer.valueOf(split[4]));
				}
			}
		);
		// 7-2.分组
		KeyedStream<Tuple2<String, Integer>, String> stream72 = stream71.keyBy(
			new KeySelector<Tuple2<String, Integer>, String>() {
				@Override
				public String getKey(Tuple2<String, Integer> value) throws Exception {
					return value.f0;
				}
			}
		);
		// 7-3.求和
		SingleOutputStreamOperator<Tuple2<String, Integer>> stream7 = stream72.sum(1);
		stream7.print();

		// todo 9、将7的计算结果写入到Mysql表中，表自行设计。（10分）
		SinkFunction<Tuple2<String, Integer>> jdbcSink = JdbcSink.sink(
			"REPLACE INTO t_stream7 (goods_category, goods_order_count) VALUES (?, ?)",
			(statement, tuple) -> {
				statement.setString(1, tuple.f0);
				statement.setInt(2, tuple.f1);
			},
			JdbcExecutionOptions.builder()
				.withBatchSize(1000)
				.withBatchIntervalMs(200)
				.withMaxRetries(5)
				.build(),
			new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
				.withUrl("jdbc:mysql://node101:3306/db_test")
				.withDriverName("com.mysql.cj.jdbc.Driver")
				.withUsername("root")
				.withPassword("123456")
				.build()
		);
		stream7.addSink(jdbcSink);

		// todo 8、根据商品类别分组,求出每类别商品下单的总金额。（10分）
		// 8-1.提取字段，封装二元组
		SingleOutputStreamOperator<Tuple2<String, Double>> stream81 = stream6.map(
			new RichMapFunction<String, Tuple2<String, Double>>() {
				@Override
				public Tuple2<String, Double> map(String value) throws Exception {
					/*
						1,玩具,玩具手枪,28888.2,12
						商品id,商品类别,商品名称,金额,商品下单个数
					 */
					// 字符串分割
					String[] split = value.split(",");
					// 商品类别、商品下单个数
					return Tuple2.of(split[1], Double.valueOf(split[3]));
				}
			}
		);
		// 8-2.分组
		KeyedStream<Tuple2<String, Double>, String> stream82 = stream81.keyBy(
			new KeySelector<Tuple2<String, Double>, String>() {
				@Override
				public String getKey(Tuple2<String, Double> value) throws Exception {
					return value.f0;
				}
			}
		);
		// 8-3.求和
		SingleOutputStreamOperator<Tuple2<String, Double>> stream8 = stream82.sum(1);
		stream8.print();

		// todo 10、将8的计算结果写入到Kafka主题中，主题自行创建。（10分）
		// 10-1. 将二元组数据转换字符串
		SingleOutputStreamOperator<String> stream10 = stream8.map(
			new RichMapFunction<Tuple2<String, Double>, String>() {
				@Override
				public String map(Tuple2<String, Double> value) throws Exception {
					return value.toString();
				}
			}
		);
		KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
			.setBootstrapServers("node101:9092,node102:9092,node103:9092")
			.setRecordSerializer(KafkaRecordSerializationSchema.builder()
				.setTopic("stream8")
				.setValueSerializationSchema(new SimpleStringSchema())
				.build()
			)
			.setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
			.build();
		stream10.sinkTo(kafkaSink);

		// 5.触发执行-execute
		env.execute("FlinkDemo02");
	}
}
